package com.netty.bfport.http.common.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.netty.bfport.http.common.es.EsUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Optional;

/**
 * @Author: zhaoyang05
 * @Data: 2022/3/24 17:29
 * @Description
 * 卡夫卡消费者
 */
@Component
@Slf4j
public class Consumer {

    @Resource
    EsUtils esUtils;

    private static volatile int NUM = 0;

    private static final String TOPIC = "healthy";

    private static final String TOPIC_GROUP1 = "FIRST_BLADE";

    @Value("${fzxy.history}")
    private Integer HISTORY;

    @Value("${fzxy.history.history.name}")
    private String HISTORY_NAME;

    @Async
    @KafkaListener(topics = TOPIC, groupId = TOPIC_GROUP1)
    public void consumer(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) throws IOException {

        Optional<?> message = Optional.ofNullable(record.value());
        if (message.isPresent()) {
            Object msg = message.get();
            JSONObject jsonObject = JSONObject.parseObject(JSON.parse(message.get().toString()).toString());
            System.out.println("状态为：：：" + NUM);
            if (NUM==HISTORY)NUM=0;
            if (NUM == 0){
                jsonObject.put(HISTORY_NAME,1);
            }else {
                jsonObject.put(HISTORY_NAME,0);
            }
            NUM++;
            jsonObject.put("createTime",new Date(jsonObject.getLong("createTime")));
            log.info("topic_test 消费了： Topic:" + topic + ",Message:" + jsonObject);
            Date date = new Date();
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
            //每天生成一个新的index，防止数据过大
            String humidity = jsonObject.getString("humidity");
            String s = humidity.replaceAll("%", "");
            jsonObject.put("humidity",s);
            esUtils.add("fz" + sdf.format(date),"user",jsonObject);
            ack.acknowledge();
        }
    }
}
